package com.nl.hotitems;


import com.nl.bean.input.UserBehavior;
import com.nl.bean.output.WindowAggResult;
import com.nl.utils.WindowAggResultOut;
import com.nl.utils.aggfunctions.CountAgg;
import java.net.URL;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


/**
 * @author shihb
 * @date 2019/12/18 17:58
 * 1小时的热门商品排名topN,每5分钟输出一次
 */
public class HotItemsMain {

  public static void main(String[] args) throws Exception {

    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


    //2.source
    URL resource = HotItemsMain.class.getClassLoader().getResource("UserBehavior.csv");
    DataStreamSource<String> source = env.readTextFile(resource.getPath());

    //3.transform算子
    SingleOutputStreamOperator<String> dataStream = source
        .map(s -> {
            // 输入数据转换成用户行为包装类
            String[] arr = s.split(",");
            long userId = Long.parseLong(arr[0].trim());
            long itemId = Long.parseLong(arr[1].trim());
            int categoryId = Integer.parseInt(arr[2].trim());
            String behavior = arr[3].trim();
            long timestamp = Long.parseLong(arr[4].trim());
            return UserBehavior.of(userId, itemId, categoryId, behavior, timestamp);
        })
        .assignTimestampsAndWatermarks(
            // 设置事件时间和水位
            new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(UserBehavior element) {
                // 要转换成毫秒
                return element.getTimestamp() * 1000;
              }
            }
        )
        // 过滤点击日志
        .filter(userBehavior -> "pv".equals(userBehavior.getBehavior()))
        // 根据商品分组,这边用到了KeySelector,保证key的类型
        .keyBy(userBehavior -> userBehavior.getItemId())
        // 设置窗口 滑窗窗口一个小时,步长5分钟
        .timeWindow(Time.hours(1), Time.minutes(5))
        // 窗口预聚合,来一条聚合一次数量,并定义结果输出
        .aggregate(new CountAgg<UserBehavior>(), new WindowAggResultOut<Long>())
        // 根据窗口分组,来统计当前窗口的topN
        .keyBy(windowResult->windowResult.getWindowEnd())
        .process(new TopHotItems(3));

    //4.sink
    dataStream.print();

    //5.执行
    env.execute("HotItems job");


  }

}

/**
 * 自定义处理函数,获取热门商品的topN
 * KeyedProcessFunction<K, I, O> key,输入,输出
 */
class TopHotItems extends KeyedProcessFunction<Long, WindowAggResult, String> {

  private int topSize;
  private ListState<WindowAggResult> itemState;

  public TopHotItems(int i) {
    this.topSize = i;
  }

  /**
   * 初始化
   * @param parameters
   */
  @Override
  public void open(Configuration parameters) {
    // 初始化状态
    itemState = getRuntimeContext().getListState(
        // 两个参数，状态名称和存储的数据类型
        new ListStateDescriptor<>("itemState", WindowAggResult.class)
    );
  }

  /**
   * 必须实现的方法，每个数据来的处理逻辑
   * @param value 输入的数据
   * @param ctx 上下文可以访问时间戳，元素的key,TimerService时间服务，侧输出流
   * @param out 处理完输出
   * @throws Exception
   */
  @Override
  public void processElement(WindowAggResult value, Context ctx, Collector<String> out)
      throws Exception {
    //每条数据来了就把状态存起来
    itemState.add(value);
    //注册定时器
    ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
  }


  /**
   * 定时器触发调用，处理定时逻辑
   * @param timestamp
   * @param ctx
   * @param out
   * @throws Exception
   */
  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
    List<WindowAggResult> allItemState = new ArrayList<>();
    // 将所有状态得数据取出
    for (WindowAggResult icv : itemState.get()) {
      allItemState.add(icv);
    }
    itemState.clear();
    // 降序排序输出
    allItemState.sort((o1,o2)->(int)(o2.getCount()-o1.getCount()));

    // 结果格式化输出
    StringBuilder sb = new StringBuilder();
    sb.append("时间:").append(new Timestamp(timestamp - 1)).append("\n");
    for (int i = 0; i < Math.min(topSize,allItemState.size()); i++) {
      WindowAggResult icv = allItemState.get(i);
      sb.append("No").append(i).append(":").append("商品ID=").append(icv.getKey()).append(" ")
          .append("浏览量=").append(icv.getCount()).append("\n");
    }
    sb.append("=========================");
    out.collect(sb.toString());
  }
}



